package org.springframework.boot.netty.sync;

import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.netty.NettyProperties;
import org.springframework.boot.netty.argumentResolver.HandlerMethodArgumentResolverComposite;
import org.springframework.boot.netty.argumentResolver.MethodParameter;
import org.springframework.boot.netty.argumentResolver.ReturnValueArgumentResolverComposite;
import org.springframework.boot.netty.argumentResolver.ThrowableResolverComposite;
import org.springframework.boot.netty.exception.ConsumerStartupException;
import org.springframework.boot.netty.filter.JwtSecurityFilter;
import org.springframework.boot.netty.listener.Message;
import org.springframework.boot.netty.listener.NettyMessageListener;
import org.springframework.core.convert.ConversionService;
import org.springframework.format.support.DefaultFormattingConversionService;
import org.springframework.util.MethodInvoker;

import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Author: huoxingzhi
 * Date: 2020/12/11
 * Email: hxz_798561819@163.com
 */
public class AsyncMessageProcessing implements Runnable {

    public static final Object lockObject = new Object();

    private final CountDownLatch start;

    private int mapIndex;

    private NettyProperties nettyProperties;

    private MethodInvoker methodInvoker;

    private Object target;

    private ConversionService conversionService;

    private HandlerMethodArgumentResolverComposite handlerMethodArgumentResolverComposite;

    private ReturnValueArgumentResolverComposite returnValueArgumentResolverComposite;

    private final SocketDataBinder socketDataBinder;

    private JwtSecurityFilter jwtSecurityFilter;

    private ThrowableResolverComposite throwableResolverComposite;

    private final static Logger logger = LoggerFactory.getLogger(NettyMessageListener.class);


    public AsyncMessageProcessing(NettyProperties nettyProperties,
                                  Object target,
                                  ConversionService conversionService,
                                  HandlerMethodArgumentResolverComposite handlerMethodArgumentResolverComposite,
                                  ReturnValueArgumentResolverComposite returnValueArgumentResolverComposite,
                                  JwtSecurityFilter jwtSecurityFilter,
                                  ThrowableResolverComposite throwableResolverComposite,
                                  int mapIndex) {
        this.nettyProperties = nettyProperties;
        this.target = target;
        this.conversionService = (conversionService == null) ? new DefaultFormattingConversionService() : conversionService;
        this.handlerMethodArgumentResolverComposite = handlerMethodArgumentResolverComposite;
        this.returnValueArgumentResolverComposite = returnValueArgumentResolverComposite;
        this.mapIndex = mapIndex;
        this.socketDataBinder = new SocketDataBinder(this.conversionService, this.handlerMethodArgumentResolverComposite);
        this.jwtSecurityFilter = jwtSecurityFilter;
        this.throwableResolverComposite = throwableResolverComposite;
        this.start = new CountDownLatch(1);
    }

    @Override
    public void run() {

        try {
            initialize();
            while (!MessageBlockQueue.shutdown) {
                mainLoop();
            }
        } catch (Exception e) {
            e.printStackTrace();
            this.start.countDown();
        }

    }


    private void initialize() {
        methodInvoker = new MethodInvoker();
        methodInvoker.setTargetObject(this.target);
        this.start.countDown();
    }

    private void mainLoop() {

            Message message = (Message) MessageBlockQueue.getMessage(this.mapIndex, this.nettyProperties.getMessageQueueTimeout(), TimeUnit.MILLISECONDS);
            Object returnValue = null;

        try {
            if (message != null && this.target != null) {
                loginIfNessary(message);
                this.socketDataBinder.doBind(message, this.target, methodInvoker);
                this.methodInvoker.prepare();
                returnValue = this.methodInvoker.invoke();
                logger.info(Thread.currentThread().getName());
                logoutIfNessary();
            } else if (message != null && this.target == null) {
                loginIfNessary(message);
                this.socketDataBinder.doBind(message, methodInvoker);
                this.methodInvoker.prepare();
                returnValue = this.methodInvoker.invoke();
                logoutIfNessary();
                logger.info(Thread.currentThread().getName());
            } else {
                logger.info(Thread.currentThread().getName());
                Thread.sleep(nettyProperties.getConsumerThreadSleep());
            }

            if (null != returnValue) {
                processReturnValue(message, returnValue, this.socketDataBinder.getBindingMethodParam());
            }

            //如果需要及时释放资源
            autoRelease(message);
       /* } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }*/
        } catch (Exception e){
            if(this.throwableResolverComposite.supportsThrowable(e)){
                this.throwableResolverComposite.resolveException(message,e);
            }else {
                e.printStackTrace();
            }

            autoRelease(message);
        }

    }

    private void loginIfNessary(Message message){
        //如果启用安全机制则校验
        if(null!=jwtSecurityFilter){
            Map<String,String> headers = new HashMap<>();
            headers.put("access_token",message.getAccessToken());
            jwtSecurityFilter.filterIfNessary(headers);
        }
    }

    private void logoutIfNessary(){
        //如果启用安全机制则校验
        if(null!=jwtSecurityFilter){
            jwtSecurityFilter.logoutIfNessary();
        }
    }

    private void autoRelease(Message message){
        if(null!=message && null!=message.getContent()){
            ReferenceCountUtil.safeRelease(message.getContent());
        }
    }


    private void processReturnValue(Message message, Object returnValue, MethodParameter methodParameter){
        if(returnValue!=null) {
            // return value 解析器处理
           if(this.returnValueArgumentResolverComposite.supportsParameter(methodParameter)){
               try {
                   this.returnValueArgumentResolverComposite.resolveArgument(message,returnValue,methodParameter);
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
        }
    }

    public ConsumerStartupException getStartupException() throws InterruptedException {
        if (!this.start.await(this.nettyProperties.getConsumerStartTimeout(), TimeUnit.MILLISECONDS)) {
            logger.error("Consumer failed to start in "
                    + this.nettyProperties.getConsumerStartTimeout()
                    + " milliseconds; does the task executor have enough threads to support the container "
                    + "concurrency?");
            return new ConsumerStartupException("Consumer failed to start in "+ this.nettyProperties.getConsumerStartTimeout());
        }
        return null;
    }
}
